package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.BaseAppV2;
import com.atguigu.gmall.realtime.bean.VisitorStats;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Map;

import static com.atguigu.gmall.realtime.common.Constant.*;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/8/31 14:36
 */
public class DwsVisitor extends BaseAppV2 {
    // 需要启动的集群:
    // 1. 日志采集  log.sh start
    // 2. 启动 DwdLog
    // 3. 启动 DwmUv
    // 4. 启动 DwmUserJumpDetail
    
    // 模拟数据
    
    public static void main(String[] args) {
        new DwsVisitor().init(4001, 1, "DwsVisitor", "DwsVisitor",
                              TOPIC_DWD_PAGE, TOPIC_DWM_UV, TOPIC_DWM_USER_JUMP_DETAIL);
    }
    
    @Override
    protected void run(StreamExecutionEnvironment env,
                       Map<String, DataStreamSource<String>> streamMap) {
    
        /*streamMap.get(TOPIC_DWD_PAGE).print(TOPIC_DWD_PAGE);
        streamMap.get(TOPIC_DWM_UV).print(TOPIC_DWM_UV);
        streamMap.get(TOPIC_DWM_USER_JUMP_DETAIL).print(TOPIC_DWM_USER_JUMP_DETAIL);*/
        
        // 1. 解析数据,并union成一个流
        DataStream<VisitorStats> visitorStatsStream = parseAndUnionOne(streamMap);
        // 2. 开窗分组聚合
        SingleOutputStreamOperator<VisitorStats> aggregatedStream = aggregate(visitorStatsStream);
        // 3. 写入到ClickHouse中
        writeToClickhouse(aggregatedStream);
        
    }
    
    private void writeToClickhouse(SingleOutputStreamOperator<VisitorStats> aggregatedStream) {
        aggregatedStream.addSink(FlinkSinkUtil.getClickhouseSink(Constant.CLICKHOUSE_DB,
                                                                 CLICKHOUSE_TABLE_VISITOR_STATS,
                                                                 VisitorStats.class));
    }
    
    private SingleOutputStreamOperator<VisitorStats> aggregate(DataStream<VisitorStats> visitorStatsStream) {
        return visitorStatsStream
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(15))
                    .withTimestampAssigner((vs, ts) -> vs.getTs())
            )
            .keyBy(vs -> vs.getVc() + ":" + vs.getCh() + ":" + vs.getAr() + ":" + vs.getIs_new())
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .sideOutputLateData(new OutputTag<VisitorStats>("late") {})
            .reduce(
                new ReduceFunction<VisitorStats>() {
                    @Override
                    public VisitorStats reduce(VisitorStats vs1,
                                               VisitorStats vs2) throws Exception {
                        vs1.setUv_ct(vs1.getUv_ct() + vs2.getUv_ct());
                        vs1.setPv_ct(vs1.getPv_ct() + vs2.getPv_ct());
                        vs1.setSv_ct(vs1.getSv_ct() + vs2.getSv_ct());
                        vs1.setUj_ct(vs1.getUj_ct() + vs2.getUj_ct());
                        vs1.setDur_sum(vs1.getDur_sum() + vs2.getDur_sum());
                        return vs1;
                        
                    }
                },
                new ProcessWindowFunction<VisitorStats, VisitorStats, String, TimeWindow>() {
                    
                    private SimpleDateFormat sdf;
                    
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    }
                    
                    @Override
                    public void process(String key,
                                        Context ctx,
                                        Iterable<VisitorStats> elements,
                                        Collector<VisitorStats> out) throws Exception {
                        VisitorStats vs = elements.iterator().next();
                        // 年月日时分秒
                        TimeWindow w = ctx.window();
                        vs.setStt(sdf.format(w.getStart()));
                        vs.setEdt(sdf.format(w.getEnd()));
                        out.collect(vs);
                    }
                }
            );
        
    }
    
    private DataStream<VisitorStats> parseAndUnionOne(Map<String, DataStreamSource<String>> streamMap) {
        // 1. 先计算PV 和持续访问时间
        SingleOutputStreamOperator<VisitorStats> pvAndDuringTimeStream = streamMap
            .get(TOPIC_DWD_PAGE)
            .map(new MapFunction<String, VisitorStats>() {
                @Override
                public VisitorStats map(String value) throws Exception {
                    JSONObject obj = JSON.parseObject(value);
                    JSONObject common = obj.getJSONObject("common");
                    String vc = common.getString("vc");
                    String ch = common.getString("ch");
                    String ar = common.getString("ar");
                    String isNew = common.getString("is_new");
                    
                    Long ts = obj.getLong("ts");
                    
                    Long durSum = obj.getJSONObject("page").getLong("during_time");
                    VisitorStats vs = new VisitorStats("", "",
                                                       vc, ch, ar, isNew,
                                                       0L, 1L, 0L, 0L, durSum,
                                                       ts);
                    return vs;
                }
            });
        
        // 2. 计算uv
        SingleOutputStreamOperator<VisitorStats> uvStream = streamMap
            .get(TOPIC_DWM_UV)
            .map(new MapFunction<String, VisitorStats>() {
                @Override
                public VisitorStats map(String value) throws Exception {
                    JSONObject obj = JSON.parseObject(value);
                    JSONObject common = obj.getJSONObject("common");
                    String vc = common.getString("vc");
                    String ch = common.getString("ch");
                    String ar = common.getString("ar");
                    String isNew = common.getString("is_new");
                    
                    Long ts = obj.getLong("ts");
                    
                    VisitorStats vs = new VisitorStats("", "",
                                                       vc, ch, ar, isNew,
                                                       1L, 0L, 0L, 0L, 0L,
                                                       ts);
                    return vs;
                }
            });
        // 3. 计算uj
        SingleOutputStreamOperator<VisitorStats> ujStream = streamMap
            .get(TOPIC_DWM_USER_JUMP_DETAIL)
            .map(new MapFunction<String, VisitorStats>() {
                @Override
                public VisitorStats map(String value) throws Exception {
                    JSONObject obj = JSON.parseObject(value);
                    JSONObject common = obj.getJSONObject("common");
                    String vc = common.getString("vc");
                    String ch = common.getString("ch");
                    String ar = common.getString("ar");
                    String isNew = common.getString("is_new");
                    
                    Long ts = obj.getLong("ts");
                    
                    VisitorStats vs = new VisitorStats("", "",
                                                       vc, ch, ar, isNew,
                                                       0L, 0L, 0L, 1L, 0L,
                                                       ts);
                    return vs;
                }
            });
        
        // 4. 计算sv
        SingleOutputStreamOperator<VisitorStats> svStream = streamMap
            .get(TOPIC_DWD_PAGE)
            .process(new ProcessFunction<String, VisitorStats>() {
                @Override
                public void processElement(String value,
                                           Context ctx,
                                           Collector<VisitorStats> out) throws Exception {
                    JSONObject obj = JSON.parseObject(value);
                    
                    String lastPageId = obj.getJSONObject("page").getString("last_page_id");
                    if (lastPageId == null || lastPageId.length() == 0) {
                        JSONObject common = obj.getJSONObject("common");
                        String vc = common.getString("vc");
                        String ch = common.getString("ch");
                        String ar = common.getString("ar");
                        String isNew = common.getString("is_new");
                        
                        Long ts = obj.getLong("ts");
                        
                        VisitorStats vs = new VisitorStats("", "",
                                                           vc, ch, ar, isNew,
                                                           0L, 0L, 1L, 0L, 0L,
                                                           ts);
                        
                        out.collect(vs);
                        
                    }
                }
            });
        
        return pvAndDuringTimeStream.union(uvStream, ujStream, svStream);
        
    }
}
